package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class Demo11DianxinData {

    private HConnection connection;
    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {
        // 1、创建连接

        Configuration configuration = new Configuration();

        /*
         * 不需要执行hmaster和regionserver的地址m  hclient会自动去zookeeper中获取，
         * 由于获取到的是一个域名，所以本地必须配置hosts
         *
         */

        //设置zookeeper地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {

            //创建连接  对数据增删改查使用
            connection = HConnectionManager.createConnection(configuration);

            /*
             * 创建修改删除表使用
             *
             */
            hBaseAdmin = new HBaseAdmin(connection);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @Test
    public void saveData() throws Exception {
        /**
         *
         *  create 'dianxin_data',{NAME => 'info', VERSIONS => 100000}
         *
         * 以 手机+天分区 号作为rowkey
         * 以数据里面的时间作为版本号
         * 以城市编号作为列名
         *
         */

        HTableInterface dianxinData = connection.getTable(TableName.valueOf("dianxin_data"));


        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/dianxin_data"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String mdn = split[0];
            String grid = split[1];
            String cityId = split[2];
            String dayId = split[7];
            String date = split[5];

            if (!"\\N".equals(date)) {

                String rowkey = mdn + "_" + dayId;

                Put put = new Put(rowkey.getBytes());

                //版本号不写默认是系统时间,  以数据时间作为版本号
                put.add("info".getBytes(), cityId.getBytes(), Long.parseLong(date), grid.getBytes());

                dianxinData.put(put);
            }
        }

    }

    @Test
    public void getTop10() throws Exception {
        /**
         *  1、通过手机号查询用户最近10条位置记录
         *
         */

        String rowkey = "FF397CE24C4DB3F4D4018C9BCE98FE549A69D1BA_20180503";

        HTableInterface dianxinData = connection.getTable(TableName.valueOf("dianxin_data"));


        Get get = new Get(rowkey.getBytes());

        //设置每个列返回版本数
        get.setMaxVersions(10);

        Result result = dianxinData.get(get);

        List<Cell> cells = result.listCells();


        ArrayList<String> list = new ArrayList<>();


        //取出所有数据
        for (Cell cell : cells) {
            long date = cell.getTimestamp();
            String city = Bytes.toString(CellUtil.cloneQualifier(cell));
            String grid = Bytes.toString(CellUtil.cloneValue(cell));

            list.add(date + "," + city + "," + grid);
        }

        //按照时间排序去前十条
        list.sort(new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
                long date1 = Long.parseLong(o1.split(",")[0]);
                long date2 = Long.parseLong(o2.split(",")[0]);
                return (int) (date2 - date1);
            }
        });


        //取前十
        List<String> top10 = list.subList(0, 10);

        for (String s : top10) {
            System.out.println(s);
        }

    }


    @Test
    public void getDayCity() throws Exception {

        /**
         * 获取用户某一天在一个城市中的所有位置
         *
         */

        String mdn = "DE9CDF69188E6C964FBA1ECADE996D08CBB1B6A9";
        String dayId = "20180503";
        String cityId = "83401";


        String rowkey = mdn + "_" + dayId;


        HTableInterface dianxinData = connection.getTable(TableName.valueOf("dianxin_data"));


        Get get = new Get(rowkey.getBytes());
        //指定查询某一列的数据
        get.addColumn("info".getBytes(), cityId.getBytes());


        //默认只查询最新一个版本的数据
        get.setMaxVersions(100000);

        Result result = dianxinData.get(get);

        List<Cell> cells = result.listCells();

        for (Cell cell : cells) {
            long date = cell.getTimestamp();
            String city = Bytes.toString(CellUtil.cloneQualifier(cell));
            String grid = Bytes.toString(CellUtil.cloneValue(cell));

            System.out.println(date + "," + city + "," + grid);
        }

    }
}
